package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"math/rand"
	"runtime"
	"sort"
	"strings"
	"sync"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// Config holds the configuration for data generation
type Config struct {
	SourceDSN          string        // 源数据库连接字符串
	TargetDSN          string        // 目标数据库连接字符串
	SchemaName         string        // 数据库模式名
	TableTypesName     string        // 各种类型测试表名
	TableCompositeName string        // 复合键测试表名
	TableEmptyName     string        // 空表表名
	RowCount           int           // 主表 (types/composite) 要生成的总行数 (百万级)
	BatchSize          int           // 批量插入的行数
	ModifyRatio        float64       // 目标库中修改行的比例 (0.0-1.0)
	DeleteRatio        float64       // 目标库中删除行的比例 (模拟源独有) (0.0-1.0)
	InsertExtraRatio   float64       // 目标库中新增行的比例 (模拟目标独有) (0.0-1.0)
	CleanupTarget      bool          // 在插入前清空目标表
	IncludeTypes       bool          // 是否包含 checksum_test_types 表
	IncludeComposite   bool          // 是否包含 checksum_test_composite_key 表
	IncludeEmpty       bool          // 是否包含 checksum_test_empty 表
	NULLProbability    float64       // 生成 NULL 值的概率 (0.0-1.0)
	Concurrency        int           // 插入数据的并行协程数量
	BatchTimeout       time.Duration // 新增: 批量插入的超时时间
}

var cfg Config

// --- 模拟中国人风格的数据组件 ---
var surnames = []string{"赵", "钱", "孙", "李", "周", "吴", "郑", "王", "冯", "陈", "褚", "卫", "蒋", "沈", "韩", "杨", "朱", "秦", "尤", "许"}
var givenNames = []string{"伟", "芳", "敏", "杰", "静", "亮", "娜", "秀英", "建华", "丽", "强", "梅", "军", "桂英", "波", "兰", "青", "军", "霞", "刚"}
var cities = []string{"北京", "上海", "广州", "深圳", "杭州", "成都", "重庆", "武汉", "苏州", "西安", "南京", "天津", "郑州", "长沙", "沈阳", "青岛"}
var streetNames = []string{"人民路", "解放街", "建设大道", "中山路", "东方广场", "世纪大道", "青年路", "幸福街", "光明路", "和平街", "团结路", "工业园"}
var jobTitles = []string{"工程师", "经理", "专员", "助理", "总监", "主管", "架构师", "设计师", "分析师", "产品经理", "销售代表", "客服"}

// 中文常用字符集 (用于生成随机中文字符串)
var chineseChars = "的一是在不了有和人这中大生用天经之去年多发工出站点部新分从动地员信理海二同应圆业产立发者规西门开市件路月足加空江何济北群制机"

// --- 表结构定义 (包含中文注释) ---
const createTableTypesSQL = `
CREATE TABLE IF NOT EXISTS %s.%s (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键ID',
    bool_col BOOLEAN COMMENT '布尔类型',
    tiny_col TINYINT COMMENT '有符号TinyInt (-128 to 127)',
    tiny_unsigned_col TINYINT UNSIGNED COMMENT '无符号TinyInt (0 to 255)',
    small_col SMALLINT COMMENT '有符号SmallInt (-32768 to 32767)',
    small_unsigned_col SMALLINT UNSIGNED COMMENT '无符号SmallInt (0 to 65535)',
    medium_col MEDIUMINT COMMENT '有符号MediumInt (-8388608 to 8388607)',
    medium_unsigned_col MEDIUMINT UNSIGNED COMMENT '无符号MediumInt (0 to 16777215)',
    int_col INT COMMENT '有符号Int (-2.1E9 to 2.1E9)',
    int_unsigned_col INT UNSIGNED COMMENT '无符号Int (0 to 4.2E9)',
    big_col BIGINT COMMENT '有符号BigInt (-9.2E18 to 9.2E18)',
    big_unsigned_col BIGINT UNSIGNED COMMENT '无符号BigInt (0 to 1.8E19)',
    decimal_col DECIMAL(10, 2) COMMENT '定点数Decimal (总共10位, 小数2位)',
    float_col FLOAT COMMENT '单精度浮点数',
    double_col DOUBLE COMMENT '双精度浮点数',
    char_col CHAR(10) COMMENT '定长字符串Char (长度10)',
    varchar_col VARCHAR(255) COMMENT '变长字符串Varchar (最大255)',
    tinytext_col TINYTEXT COMMENT '微小文本TinyText (最大255)',
    text_col TEXT COMMENT '普通文本Text (最大64KB)',
    mediumtext_col MEDIUMTEXT COMMENT '中等文本MediumText (最大16MB)',
    longtext_col LONGTEXT COMMENT '长文本LongText (最大4GB)',
    binary_col BINARY(16) COMMENT '定长二进制Binary (长度16)',
    varbinary_col VARBINARY(255) COMMENT '变长二进制Varbinary (最大255)',
    tinyblob_col TINYBLOB COMMENT '微小二进制TinyBlob (最大255)',
    blob_col BLOB COMMENT '普通二进制Blob (最大64KB)',
    mediumblob_col MEDIUMBLOB COMMENT '中等二进制MediumBlob (最大16MB)',
    longblob_col LONGBLOB COMMENT '长二进制LongBlob (最大4GB)',
    date_col DATE COMMENT '日期类型 (YYYY-MM-DD)',
    time_col TIME COMMENT '时间类型 (HH:MM:SS)',
    datetime_col DATETIME COMMENT '日期时间类型 (YYYY-MM-DD HH:MM:SS)',
    datetime_col_was_ts DATETIME NULL DEFAULT NULL COMMENT '模拟原Timestamp列,允许NULL (YYYY-MM-DD HH:MM:SS.ffffff)',
    year_col YEAR COMMENT '年份类型 (1901-2155)',
    enum_col ENUM('apple', 'banana', 'orange') COMMENT '枚举类型',
    set_col SET('red', 'green', 'blue') COMMENT '集合类型',
    json_col JSON COMMENT 'JSON类型',
    nullable_int INT NULL COMMENT '可空Int',
    nullable_varchar VARCHAR(50) NULL COMMENT '可空Varchar (最大50)',
    nullable_text TEXT NULL COMMENT '可空Text',
    nullable_datetime DATETIME NULL COMMENT '可空Datetime',
    nullable_blob BLOB NULL COMMENT '可空Blob',
    PRIMARY KEY (id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT '测试各种数据类型';
`

const createTableCompositeKeySQL = `
CREATE TABLE IF NOT EXISTS %s.%s (
    user_id INT UNSIGNED NOT NULL COMMENT '用户ID',
    item_id INT UNSIGNED NOT NULL COMMENT '商品ID',
    description VARCHAR(100) COMMENT '商品描述',
    quantity INT COMMENT '购买数量',
    last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最后更新时间戳',
    PRIMARY KEY (user_id, item_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT '测试复合主键';
`

const createTableEmptySQL = `
CREATE TABLE IF NOT EXISTS %s.%s (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键ID',
    name VARCHAR(50) COMMENT '名称',
    PRIMARY KEY (id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT '测试空表';
`

func main() {
	// 解析命令行参数
	flag.StringVar(&cfg.SourceDSN, "source-dsn", "root:123456@tcp(127.0.0.1:3306)/checksum_test_db?charset=utf8mb4&collation=utf8mb4_unicode_ci&parseTime=true", "源数据库连接字符串")
	flag.StringVar(&cfg.TargetDSN, "target-dsn", "root:123456@tcp(127.0.0.1:3307)/checksum_test_db?charset=utf8mb4&collation=utf8mb4_unicode_ci&parseTime=true", "目标数据库连接字符串")
	flag.StringVar(&cfg.SchemaName, "schema-name", "checksum_test_db", "数据库模式名")
	flag.StringVar(&cfg.TableTypesName, "table-types-name", "checksum_test_types", "各种类型测试表名")
	flag.StringVar(&cfg.TableCompositeName, "table-composite-name", "checksum_test_composite_key", "复合键测试表名")
	flag.StringVar(&cfg.TableEmptyName, "table-empty-name", "checksum_test_empty", "空表表名")
	flag.IntVar(&cfg.RowCount, "row-count", 10000, "主表 (types/composite) 要生成的总行数")
	flag.IntVar(&cfg.BatchSize, "batch-size", 1000, "批量插入的行数 (已调整，避免超过MySQL占位符限制)")
	flag.Float64Var(&cfg.ModifyRatio, "modify-ratio", 0.002, "目标库中修改行的比例 (0.0-1.0)")
	flag.Float64Var(&cfg.DeleteRatio, "delete-ratio", 0.001, "目标库中删除行的比例 (模拟源独有) (0.0-1.0)")
	flag.Float64Var(&cfg.InsertExtraRatio, "insert-extra-ratio", 0.001, "目标库中新增行的比例 (模拟目标独有) (0.0-1.0)")
	flag.BoolVar(&cfg.CleanupTarget, "cleanup-target", true, "插入前清空目标表")
	flag.BoolVar(&cfg.IncludeTypes, "include-types", true, "包含 checksum_test_types 表")
	flag.BoolVar(&cfg.IncludeComposite, "include-composite", true, "包含 checksum_test_composite_key 表")
	flag.BoolVar(&cfg.IncludeEmpty, "include-empty", true, "包含 checksum_test_empty 表")
	flag.Float64Var(&cfg.NULLProbability, "null-prob", 0.1, "生成 NULL 值的概率 (0.0-1.0)")
	flag.IntVar(&cfg.Concurrency, "concurrency", runtime.NumCPU(), "插入数据的并行协程数量 (默认为CPU核心数)")
	flag.DurationVar(&cfg.BatchTimeout, "batch-timeout", 30*time.Second, "批量插入的超时时间") // 新增超时参数

	flag.Parse()

	// 初始化随机数种子，使用当前时间
	rand.Seed(time.Now().UnixNano())

	log.SetFlags(log.LstdFlags | log.Lshortfile)

	// --- 连接数据库 ---
	log.Println("正在连接源数据库...")
	sourceDB, err := sql.Open("mysql", cfg.SourceDSN)
	if err != nil {
		log.Fatalf("连接源数据库失败: %v", err)
	}
	defer sourceDB.Close()
	// 设置连接池参数
	sourceDB.SetMaxOpenConns(cfg.Concurrency + 10) // 适当增加连接池大小
	sourceDB.SetMaxIdleConns(cfg.Concurrency + 5)
	sourceDB.SetConnMaxLifetime(time.Hour)

	if err := sourceDB.Ping(); err != nil {
		log.Fatalf("Ping 源数据库失败: %v", err)
	}
	log.Println("源数据库连接成功。")

	log.Println("正在连接目标数据库...")
	targetDB, err := sql.Open("mysql", cfg.TargetDSN)
	if err != nil {
		log.Fatalf("连接目标数据库失败: %v", err)
	}
	defer targetDB.Close()
	// 设置连接池参数
	targetDB.SetMaxOpenConns(cfg.Concurrency + 10) // 适当增加连接池大小
	targetDB.SetMaxIdleConns(cfg.Concurrency + 5)
	targetDB.SetConnMaxLifetime(time.Hour)

	if err := targetDB.Ping(); err != nil {
		log.Fatalf("Ping 目标数据库失败: %v", err)
	}
	log.Println("目标数据库连接成功。")

	ctx := context.Background()

	// --- 创建表 ---
	log.Println("正在创建测试表...")
	if cfg.IncludeTypes {
		if _, err := sourceDB.ExecContext(ctx, fmt.Sprintf(createTableTypesSQL, cfg.SchemaName, cfg.TableTypesName)); err != nil {
			log.Fatalf("创建源库 %s 表失败: %v", cfg.TableTypesName, err)
		}
		if _, err := targetDB.ExecContext(ctx, fmt.Sprintf(createTableTypesSQL, cfg.SchemaName, cfg.TableTypesName)); err != nil {
			log.Fatalf("创建目标库 %s 表失败: %v", cfg.TableTypesName, err)
		}
		log.Printf("表 %s.%s 已创建。", cfg.SchemaName, cfg.TableTypesName)
	}
	if cfg.IncludeComposite {
		if _, err := sourceDB.ExecContext(ctx, fmt.Sprintf(createTableCompositeKeySQL, cfg.SchemaName, cfg.TableCompositeName)); err != nil {
			log.Fatalf("创建源库 %s 表失败: %v", cfg.TableCompositeName, err)
		}
		if _, err := targetDB.ExecContext(ctx, fmt.Sprintf(createTableCompositeKeySQL, cfg.SchemaName, cfg.TableCompositeName)); err != nil {
			log.Fatalf("创建目标库 %s 表失败: %v", cfg.TableCompositeName, err)
		}
		log.Printf("表 %s.%s 已创建。", cfg.SchemaName, cfg.TableCompositeName)
	}
	if cfg.IncludeEmpty {
		// 确保空表在源和目标都存在且为空
		dropEmptySQL := fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", cfg.SchemaName, cfg.TableEmptyName)
		if _, err := sourceDB.ExecContext(ctx, dropEmptySQL); err != nil {
			log.Fatalf("删除源库旧空表失败: %v", err)
		}
		if _, err := targetDB.ExecContext(ctx, dropEmptySQL); err != nil {
			log.Fatalf("删除目标库旧空表失败: %v", err)
		}
		if _, err := sourceDB.ExecContext(ctx, fmt.Sprintf(createTableEmptySQL, cfg.SchemaName, cfg.TableEmptyName)); err != nil {
			log.Fatalf("创建源库空表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableEmptyName, err)
		}
		if _, err := targetDB.ExecContext(ctx, fmt.Sprintf(createTableEmptySQL, cfg.SchemaName, cfg.TableEmptyName)); err != nil {
			log.Fatalf("创建目标库空表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableEmptyName, err)
		}
		log.Printf("表 %s.%s (空表) 已创建。", cfg.SchemaName, cfg.TableEmptyName)
	}
	log.Println("测试表创建完成。")

	// --- 清空目标表数据 ---
	log.Printf("正在清空目标表数据...")

	// 先清空源库表
	if cfg.IncludeTypes {
		_, err = sourceDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableTypesName))
		if err != nil {
			log.Fatalf("清空表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableTypesName, err)
		}
		log.Printf("表 %s.%s 已清空。", cfg.SchemaName, cfg.TableTypesName)
	}

	if cfg.IncludeComposite {
		_, err = sourceDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableCompositeName))
		if err != nil {
			log.Fatalf("清空表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableCompositeName, err)
		}
		log.Printf("表 %s.%s 已清空。", cfg.SchemaName, cfg.TableCompositeName)
	}

	// 再清空目标库表
	if cfg.IncludeTypes {
		_, err = targetDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableTypesName))
		if err != nil {
			log.Fatalf("清空表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableTypesName, err)
		}
		log.Printf("表 %s.%s 已清空。", cfg.SchemaName, cfg.TableTypesName)
	}

	if cfg.IncludeComposite {
		_, err = targetDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableCompositeName))
		if err != nil {
			log.Fatalf("清空表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableCompositeName, err)
		}
		log.Printf("表 %s.%s 已清空。", cfg.SchemaName, cfg.TableCompositeName)
	}

	log.Printf("目标表清空完成。")

	// --- 生成并插入数据 ---
	log.Printf("开始生成并插入 %d 行数据...", cfg.RowCount)
	if cfg.IncludeTypes {
		log.Printf("正在处理表: %s.%s", cfg.SchemaName, cfg.TableTypesName)
		if err := generateAndInsertData(ctx, sourceDB, targetDB, cfg.SchemaName, cfg.TableTypesName, cfg.RowCount, cfg.BatchSize, cfg.ModifyRatio, cfg.DeleteRatio, cfg.InsertExtraRatio, generateTypesRow, 1); err != nil {
			log.Fatalf("生成并插入数据到表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableTypesName, err)
		}
		log.Printf("表 %s.%s 数据生成完成。", cfg.SchemaName, cfg.TableTypesName)
	}

	if cfg.IncludeComposite {
		log.Printf("正在处理表: %s.%s", cfg.SchemaName, cfg.TableCompositeName)
		if err := generateAndInsertData(ctx, sourceDB, targetDB, cfg.SchemaName, cfg.TableCompositeName, cfg.RowCount, cfg.BatchSize, cfg.ModifyRatio, cfg.DeleteRatio, cfg.InsertExtraRatio, generateCompositeRow, 1); err != nil {
			log.Fatalf("生成并插入数据到表 %s.%s 失败: %v", cfg.SchemaName, cfg.TableCompositeName, err)
		}
		log.Printf("表 %s.%s 数据生成完成。", cfg.SchemaName, cfg.TableCompositeName)
	}

	if cfg.IncludeEmpty {
		log.Printf("表 %s.%s 被设置为保持为空。", cfg.SchemaName, cfg.TableEmptyName)
	}

	log.Println("\n--- 所有测试数据生成完成 ---")
	log.Printf("您现在可以使用校验工具对 %s.%s 和 %s.%s 进行性能和功能测试。", cfg.SchemaName, cfg.TableTypesName, cfg.SchemaName, cfg.TableCompositeName)
}

// generateAndInsertData 生成并插入数据到源库和目标库
// 此函数协调生成器、批处理器和插入器协程
func generateAndInsertData(ctx context.Context, sourceDB, targetDB *sql.DB, schema, table string, rowCount int, batchSize int, modifyRatio, deleteRatio, insertExtraRatio float64, generateRowFunc func(int, string, int) map[string]interface{}, concurrency int) error {

	// --- 唯一组合生成逻辑，仅对 composite_key 表生效 ---
	var compositeCombos [][2]uint32
	var extraCombos [][2]uint32
	isComposite := (table == cfg.TableCompositeName)
	// itemRangePerUser := 3000
	if isComposite {
		// 用 map 保证组合绝对唯一
		comboSet := make(map[[2]uint32]struct{})
		compositeCombos = compositeCombos[:0]
		for len(compositeCombos) < rowCount {
			userID := uint32(1)
			itemID := uint32(len(compositeCombos) + 1)
			// 保证 user_id/item_id 递增且唯一
			combo := [2]uint32{userID, itemID}
			if _, exists := comboSet[combo]; !exists {
				comboSet[combo] = struct{}{}
				compositeCombos = append(compositeCombos, combo)
			}
		}
		// 检查池唯一性
		log.Printf("【调试】compositeCombos 池大小: %d", len(compositeCombos))
		comboCheck := make(map[string]struct{})
		for i, c := range compositeCombos {
			key := fmt.Sprintf("%d-%d", c[0], c[1])
			if _, exists := comboCheck[key]; exists {
				log.Fatalf("【致命错误】compositeCombos 池中有重复组合: %s", key)
			}
			comboCheck[key] = struct{}{}
			if i < 10 {
				log.Printf("【调试】compositeCombos[%d] = (%d,%d)", i, c[0], c[1])
			}
		}
		// 确认表已清空
		var count int
		err := sourceDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s.%s", schema, table)).Scan(&count)
		if err != nil {
			log.Printf("【警告】检查表 %s.%s 是否为空失败: %v", schema, table, err)
		} else if count > 0 {
			log.Fatalf("【致命错误】表 %s.%s 在插入前未清空，当前有 %d 行数据", schema, table, count)
		} else {
			log.Printf("【调试】表 %s.%s 已确认为空", schema, table)
		}
		// 打乱顺序
		rand.Shuffle(len(compositeCombos), func(i, j int) {
			compositeCombos[i], compositeCombos[j] = compositeCombos[j], compositeCombos[i]
		})
		// 再次检查池唯一性（shuffle后）
		comboCheck = make(map[string]struct{})
		for i, c := range compositeCombos {
			key := fmt.Sprintf("%d-%d", c[0], c[1])
			if _, exists := comboCheck[key]; exists {
				log.Fatalf("【致命错误】shuffle后 compositeCombos 池中有重复组合: %s", key)
			}
			comboCheck[key] = struct{}{}
			if i < 10 {
				log.Printf("【调试】shuffle后 compositeCombos[%d] = (%d,%d)", i, c[0], c[1])
			}
		}
	}

	// 生成 extra 组合池
	extraUserOffset := uint32(100000)
	extraCount := int(float64(rowCount) * insertExtraRatio)
	extraCombos = extraCombos[:0]
	comboSetExtra := make(map[[2]uint32]struct{})
	for len(extraCombos) < extraCount {
		userID := uint32(1 + extraUserOffset)
		itemID := uint32(len(extraCombos) + 1)
		combo := [2]uint32{userID, itemID}
		if _, exists := comboSetExtra[combo]; !exists {
			comboSetExtra[combo] = struct{}{}
			extraCombos = append(extraCombos, combo)
		}
	}
	// 检查 extra 池唯一性
	log.Printf("【调试】extraCombos 池大小: %d", len(extraCombos))
	extraComboCheck := make(map[string]struct{})
	for i, c := range extraCombos {
		key := fmt.Sprintf("%d-%d", c[0], c[1])
		if _, exists := extraComboCheck[key]; exists {
			log.Fatalf("【致命错误】extraCombos 池中有重复组合: %s", key)
		}
		extraComboCheck[key] = struct{}{}
		if i < 5 {
			log.Printf("【调试】extraCombos[%d] = (%d,%d)", i, c[0], c[1])
		}
	}
	// 打乱 extra 组合顺序
	rand.Shuffle(len(extraCombos), func(i, j int) {
		extraCombos[i], extraCombos[j] = extraCombos[j], extraCombos[i]
	})

	// --- 源库阶段 ---
	log.Printf("源库阶段: 开始处理表 %s.%s (生成 %d 行)...", schema, table, rowCount)
	sourceCtx, sourceCancel := context.WithCancel(ctx)
	defer sourceCancel()

	sourceRowChan := make(chan map[string]interface{}, batchSize*2)
	sourceBatchChan := make(chan []map[string]interface{}, concurrency*2)
	errChan := make(chan error, concurrency+3)

	var wg sync.WaitGroup

	// 1. 源库生成器
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(sourceRowChan)
		log.Printf("源库生成器启动...")
		if isComposite {
			// 检查源库生成器是否被多次启动
			log.Printf("【调试】源库生成器启动，将生成 %d 行数据", rowCount)
			// 记录已用组合，确保不重复
			usedCombos := make(map[string]struct{})
			for i := 0; i < rowCount; i++ {
				userID := compositeCombos[i][0]
				itemID := compositeCombos[i][1]
				key := fmt.Sprintf("%d-%d", userID, itemID)
				if _, exists := usedCombos[key]; exists {
					log.Fatalf("【致命错误】源库生成器即将插入重复组合: %s (索引 %d)", key, i)
				}
				usedCombos[key] = struct{}{}
				row := generateCompositeRowWithKey(userID, itemID, "source", rowCount)
				sourceRowChan <- row
				if i < 10 || (i > 0 && i%100000 == 0) {
					log.Printf("【调试】源库生成第 %d 行: (%d,%d)", i, userID, itemID)
				}
			}
			log.Printf("【调试】源库生成器完成，共生成 %d 行，所有组合均唯一", len(usedCombos))
		} else {
			for i := 0; i < rowCount; i++ {
				select {
				case <-sourceCtx.Done():
					log.Printf("源库生成器收到取消信号，退出。")
					return
				default:
					// continue
				}
				sourceRowChan <- generateRowFunc(i, "source", rowCount)
				if i > 0 && i%100000 == 0 {
					log.Printf("源库生成器已生成 %d 行...", i)
				}
			}
			log.Printf("源库生成器完成，共生成 %d 行。", rowCount)
		}
	}()

	// 2. 源库批处理器
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(sourceBatchChan)
		log.Printf("源库批处理器启动...")
		// 记录批处理器已处理的组合
		batchProcessedCombos := make(map[string]struct{})
		batch := make([]map[string]interface{}, 0, batchSize)
		count := 0
		for row := range sourceRowChan {
			select {
			case <-sourceCtx.Done():
				log.Printf("源库批处理器收到取消信号，退出。")
				return
			default:
				// continue
			}
			batch = append(batch, row)
			if isComposite {
				// 检查批处理器是否收到重复组合
				key := fmt.Sprintf("%d-%d", row["user_id"], row["item_id"])
				if _, exists := batchProcessedCombos[key]; exists {
					log.Fatalf("【致命错误】批处理器收到重复组合: %s (已处理 %d 行)", key, count)
				}
				batchProcessedCombos[key] = struct{}{}
				if count < 10 {
					log.Printf("【调试】批处理器处理第 %d 行: (%d,%d)", count, row["user_id"], row["item_id"])
				}
			}
			if len(batch) == batchSize {
				count += len(batch)
				log.Printf("源库批处理器 #%d 尝试发送满批次 (%d 行), 总计已处理 %d 行...", 0, batchSize, count)
				select {
				case sourceBatchChan <- batch:
					log.Printf("源库批处理器 #%d 成功发送满批次。", 0)
					batch = make([]map[string]interface{}, 0, batchSize)
				case <-sourceCtx.Done():
					log.Printf("源库批处理器 #%d 发送批量时收到取消信号，退出。", 0)
					return
				}
			}
		}
		log.Printf("源库批处理器 #%d 已处理完所有单行数据。总计处理行数 (发送到批次) %d.", 0, count)

		if len(batch) > 0 {
			count += len(batch)
			log.Printf("源库批处理器 #0 尝试发送最后一个批次 (%d 行), 总计已处理 %d 行...", len(batch), count)
			select {
			case sourceBatchChan <- batch:
				log.Printf("源库批处理器 #0 成功发送最后一个批次。", 0)
			case <-sourceCtx.Done():
				log.Printf("源库批处理器 #0 发送最后一个批次时收到取消信号，退出。", 0)
				return
			}
		}
		log.Printf("源库批处理器 #0 完成并准备关闭通道。")
	}()

	// 3. 源库插入器池（修正：插入器数量不超过批次数，且每个批次只被一个插入器消费）
	batchCount := rowCount / batchSize
	if rowCount%batchSize != 0 {
		batchCount++
	}
	workerCount := batchCount
	if workerCount > 1 {
		workerCount = 1 // 强制只启动1个插入器，彻底避免并发主键冲突
	}
	log.Printf("启动 %d 个源库插入器协程...", workerCount)
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			stageName := "源库"
			log.Printf("%s 插入器 #%d 启动...", stageName, workerID)
			for batch := range sourceBatchChan {
				select {
				case <-sourceCtx.Done():
					log.Printf("%s 插入器 #%d 收到取消信号，退出。", stageName, workerID)
					return
				default:
					batchCtx, cancelBatch := context.WithTimeout(sourceCtx, cfg.BatchTimeout)
					err := insertRowsInBatches(batchCtx, sourceDB, schema, table, batch, batchSize)
					cancelBatch()
					if err != nil {
						log.Printf("%s 插入器 #%d 插入批量失败 (%d 行): %v", stageName, workerID, len(batch), err)
						select {
						case errChan <- fmt.Errorf("%s 插入失败: %w", stageName, err):
						case <-sourceCtx.Done():
						}
						sourceCancel()
						return
					}
				}
			}
			log.Printf("%s 插入器 #%d 完成。", stageName, workerID)
		}(i)
	}

	go func() {
		wg.Wait()
		close(errChan)
		log.Printf("源库阶段所有协程已完成或退出，错误通道已关闭。")
	}()

	if err := <-errChan; err != nil {
		return fmt.Errorf("源库数据生成或插入过程中发生错误: %w", err)
	}
	log.Printf("源库阶段完成。")

	// --- 目标库阶段 ---
	log.Printf("目标库阶段: 开始处理表 %s.%s (生成 %d 行基础 + 差异)...", schema, table, rowCount)
	targetCtx, targetCancel := context.WithCancel(ctx)
	defer targetCancel()

	targetRowChan := make(chan map[string]interface{}, batchSize*2)
	targetBatchChan := make(chan []map[string]interface{}, concurrency*2)
	targetErrChan := make(chan error, concurrency+3)

	var targetWg sync.WaitGroup

	// 1. 目标库生成器
	targetWg.Add(1)
	go func() {
		defer targetWg.Done()
		defer close(targetRowChan)
		log.Printf("目标库生成器启动...")

		if isComposite {
			totalTargetRowsToGenerate := rowCount + int(float64(rowCount)*insertExtraRatio)
			numModified := 0
			numDeleted := 0
			numExtraInserted := 0
			modCombos := make([][2]uint32, len(compositeCombos))
			copy(modCombos, compositeCombos)
			rand.Shuffle(len(modCombos), func(i, j int) {
				modCombos[i], modCombos[j] = modCombos[j], modCombos[i]
			})
			extraUsed := 0
			// Add a map to track all used combinations in the target database
			targetUsedCombos := make(map[string]struct{})
			for i := 0; i < totalTargetRowsToGenerate; i++ {
				select {
				case <-targetCtx.Done():
					log.Printf("目标库生成器收到取消信号，退出。")
					return
				default:
					// continue
				}
				// 决定当前行是否为源库中被删除的行
				isDeleted := false
				if i < rowCount && rand.Float64() < deleteRatio {
					numDeleted++
					isDeleted = true
				}
				if !isDeleted {
					var generatedRow map[string]interface{}
					if i < rowCount && rand.Float64() < modifyRatio {
						userID := modCombos[i][0]
						itemID := modCombos[i][1]
						// Check if this combination has already been used
						key := fmt.Sprintf("%d-%d", userID, itemID)
						if _, exists := targetUsedCombos[key]; exists {
							// Skip this combination and try another one
							continue
						}
						targetUsedCombos[key] = struct{}{}
						generatedRow = generateCompositeRowWithKey(userID, itemID, "modified", rowCount)
						numModified++
					} else if i < rowCount {
						userID := compositeCombos[i][0]
						itemID := compositeCombos[i][1]
						// Check if this combination has already been used
						key := fmt.Sprintf("%d-%d", userID, itemID)
						if _, exists := targetUsedCombos[key]; exists {
							// Skip this combination and try another one
							continue
						}
						targetUsedCombos[key] = struct{}{}
						generatedRow = generateCompositeRowWithKey(userID, itemID, "source", rowCount)
					} else {
						if extraUsed < len(extraCombos) {
							userID := extraCombos[extraUsed][0]
							itemID := extraCombos[extraUsed][1]
							generatedRow = generateCompositeRowWithKey(userID, itemID, "extra", rowCount)
							extraUsed++
							numExtraInserted++
						}
					}
					if generatedRow != nil {
						targetRowChan <- generatedRow
					}
				}
				if i > 0 && i%100000 == 0 {
					log.Printf("目标库生成器已生成 %d 行...", i)
				}
			}
			log.Printf("目标库生成器完成。计划处理 %d 行 (源库 %d 行基准: 跳过 %d, 修改 %d; 额外新增 %d)。实际发送到通道 %d 行。",
				totalTargetRowsToGenerate, rowCount, numDeleted, numModified, numExtraInserted, totalTargetRowsToGenerate-numDeleted)
		} else {
			// 非 composite 表逻辑保持不变
			totalTargetRowsToGenerate := rowCount + int(float64(rowCount)*insertExtraRatio)
			numModified := 0
			numDeleted := 0
			numExtraInserted := 0
			for i := 0; i < totalTargetRowsToGenerate; i++ {
				select {
				case <-targetCtx.Done():
					log.Printf("目标库生成器收到取消信号，退出。")
					return
				default:
					// continue
				}
				isDeleted := false
				if i < rowCount && rand.Float64() < deleteRatio {
					numDeleted++
					isDeleted = true
				}
				if !isDeleted {
					var generatedRow map[string]interface{}
					if i < rowCount && rand.Float64() < modifyRatio {
						generatedRow = generateRowFunc(i, "modified", rowCount)
						numModified++
					} else if i < rowCount {
						generatedRow = generateRowFunc(i, "source", rowCount)
					} else {
						generatedRow = generateRowFunc(i, "extra", rowCount)
						numExtraInserted++
					}
					if generatedRow != nil {
						targetRowChan <- generatedRow
					}
				}
				if i > 0 && i%100000 == 0 {
					log.Printf("目标库生成器已生成 %d 行...", i)
				}
			}
			log.Printf("目标库生成器完成。计划处理 %d 行 (源库 %d 行基准: 跳过 %d, 修改 %d; 额外新增 %d)。实际发送到通道 %d 行。",
				totalTargetRowsToGenerate, rowCount, numDeleted, numModified, numExtraInserted, totalTargetRowsToGenerate-numDeleted)
		}
	}()

	// 2. 目标库批处理器
	targetWg.Add(1)
	go func() {
		defer targetWg.Done()
		defer close(targetBatchChan)
		log.Printf("目标库批处理器启动...")
		// 记录批处理器已处理的组合
		batchProcessedCombos := make(map[string]struct{})
		batch := make([]map[string]interface{}, 0, batchSize)
		count := 0
		for row := range targetRowChan {
			if isComposite {
				// 检查批处理器是否收到重复组合
				key := fmt.Sprintf("%d-%d", row["user_id"], row["item_id"])
				if _, exists := batchProcessedCombos[key]; exists {
					log.Fatalf("【致命错误】批处理器收到重复组合: %s (已处理 %d 行)", key, count)
				}
				batchProcessedCombos[key] = struct{}{}
				if count < 10 {
					log.Printf("【调试】批处理器处理第 %d 行: (%d,%d)", count, row["user_id"], row["item_id"])
				}
			}
			batch = append(batch, row)
			if len(batch) == batchSize {
				count += len(batch)
				log.Printf("目标库批处理器 #%d 尝试发送满批次 (%d 行), 总计已处理 %d 行...", 0, batchSize, count)
				select {
				case targetBatchChan <- batch:
					log.Printf("目标库批处理器 #%d 成功发送满批次。", 0)
					batch = make([]map[string]interface{}, 0, batchSize)
				case <-targetCtx.Done():
					log.Printf("目标库批处理器 #%d 发送批量时收到取消信号，退出。", 0)
					return
				}
			}
		}
		log.Printf("目标库批处理器 #%d 已处理完所有单行数据。总计处理行数 (发送到批次) %d.", 0, count)

		if len(batch) > 0 {
			count += len(batch)
			log.Printf("目标库批处理器 #0 尝试发送最后一个批次 (%d 行), 总计已处理 %d 行...", len(batch), count)
			select {
			case targetBatchChan <- batch:
				log.Printf("目标库批处理器 #0 成功发送最后一个批次。", 0)
			case <-targetCtx.Done():
				log.Printf("目标库批处理器 #0 发送最后一个批次时收到取消信号，退出。", 0)
				return
			}
		}
		log.Printf("目标库批处理器 #0 完成并准备关闭通道。")
	}()

	// 3. 目标库插入器池（同样只启动1个插入器）
	log.Printf("启动 1 个目标库插入器协程...")
	targetWg.Add(1)
	go func() {
		defer targetWg.Done()
		stageName := "目标库"
		log.Printf("%s 插入器 #0 启动...", stageName)
		for batch := range targetBatchChan {
			select {
			case <-targetCtx.Done():
				log.Printf("%s 插入器 #0 收到取消信号，退出。", stageName)
				return
			default:
				batchCtx, cancelBatch := context.WithTimeout(targetCtx, cfg.BatchTimeout)
				err := insertRowsInBatches(batchCtx, targetDB, schema, table, batch, batchSize)
				cancelBatch()
				if err != nil {
					log.Printf("%s 插入器 #0 插入批量失败 (%d 行): %v", stageName, len(batch), err)
					select {
					case targetErrChan <- fmt.Errorf("%s 插入失败: %w", stageName, err):
					case <-targetCtx.Done():
					}
					targetCancel()
					return
				}
			}
		}
		log.Printf("%s 插入器 #0 完成。", stageName)
	}()

	go func() {
		targetWg.Wait()
		close(targetErrChan)
		log.Printf("目标库阶段所有协程已完成或退出，错误通道已关闭。")
	}()

	if err := <-targetErrChan; err != nil {
		return fmt.Errorf("目标库数据生成或插入过程中发生错误: %w", err)
	}
	log.Printf("目标库阶段完成。")

	return nil
}

// insertRowsInBatches 将行数据批量插入到指定表
// ctx: 上下文 (现在带有超时)
// db: 数据库连接池
// schema, table: 模式名和表名
// rows: 要插入的当前批次行数据切片
// batchSize: 批量大小 (仅用于报告，实际批量大小是 len(rows))
func insertRowsInBatches(ctx context.Context, db *sql.DB, schema, table string, rows []map[string]interface{}, batchSize int) error {
	if len(rows) == 0 {
		return nil
	}

	// 从第一行获取所有列名，并排序
	allColumns := make([]string, 0, len(rows[0]))
	for col := range rows[0] {
		allColumns = append(allColumns, col)
	}
	sort.Strings(allColumns) // 对列名进行排序

	columnsToInsert := make([]string, 0, len(allColumns))
	// 预估总值数量 len(rows) * len(columnsToInsert)
	// 如果 types 表跳过 id， columnsToInsert 长度会比 allColumns 少1
	isTypesTable := (table == cfg.TableTypesName)
	estimatedColsPer但这Batch := len(allColumns)
	if isTypesTable {
		estimatedColsPer但这Batch--
	}
	valsToInsert := make([]interface{}, 0, len(rows)*estimatedColsPer但这Batch)

	// 确定要插入的列名列表
	for _, col := range allColumns {
		if isTypesTable && col == "id" {
			continue // 如果是 types 表，跳过 id 列，让数据库自动生成
		}
		columnsToInsert = append(columnsToInsert, col)
	}

	// 构建 INSERT 语句的列部分和单行 VALUES 占位符
	quotedCols := make([]string, len(columnsToInsert))
	placeholdersPerRow := make([]string, len(columnsToInsert))
	for i, col := range columnsToInsert {
		quotedCols[i] = fmt.Sprintf("`%s`", col)
		placeholdersPerRow[i] = "?"
	}
	colsSQL := strings.Join(quotedCols, ", ")
	placeholderRowSQL := "(" + strings.Join(placeholdersPerRow, ", ") + ")" // 单行占位符 (?, ?, ...)

	// 构建当前批次的 VALUES 部分和参数切片
	var valuePlaceholders []string
	for _, row := range rows { // Iterate over the current batch
		valuePlaceholders = append(valuePlaceholders, placeholderRowSQL)
		// 按照 columnsToInsert 的顺序提取值
		for _, col := range columnsToInsert {
			// 从原始 row map 中获取对应列的值
			// 检查是否存在，理论上应存在，但以防万一
			val, ok := row[col]
			if !ok {
				// This should not happen if generateRowFunc is consistent
				log.Printf("警告: 行数据缺少列 '%s'", col)
				valsToInsert = append(valsToInsert, nil) // Append nil or appropriate zero value? nil is safer for dynamic types
			} else {
				valsToInsert = append(valsToInsert, val)
			}
		}
	}
	valuesSQL := strings.Join(valuePlaceholders, ", ")

	baseInsertSQL := fmt.Sprintf("INSERT INTO `%s`.`%s` (%s) VALUES ", schema, table, colsSQL)
	batchInsertSQL := baseInsertSQL + valuesSQL

	// 在事务中执行批量插入
	// 使用带上下文的 BeginTx
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		// 如果 context 已经被取消，BeginTx 会返回 context.Canceled 或 context.DeadlineExceeded
		// 如果是其他错误，可能是数据库连接问题
		if ctx.Err() != nil {
			return fmt.Errorf("在 context 完成时开启事务失败: %w", err)
		}
		return fmt.Errorf("开启事务失败: %w", err)
	}

	// 使用带上下文的 ExecContext
	// ExecContext 应该会尊重 ctx 的超时或取消
	if _, err := tx.ExecContext(ctx, batchInsertSQL, valsToInsert...); err != nil {
		// Rollback the transaction on error
		if rbErr := tx.Rollback(); rbErr != nil {
			log.Printf("回滚事务失败: %v", rbErr)
		}
		// 记录错误信息
		// log.Printf("执行批量插入失败 (%s.%s 批次 %d 行): %v", schema, table, len(rows), err) // Moved logging to caller
		// 返回主错误
		return fmt.Errorf("执行批量插入失败: %w", err)
	}

	// Commit the transaction on success
	if err := tx.Commit(); err != nil {
		// log.Printf("提交事务失败 (%s.%s 批次 %d 行): %v", schema, table, len(rows), err) // Moved logging to caller
		return fmt.Errorf("提交事务失败: %w", err)
	}

	return nil
}

// --- 随机数据生成基础辅助函数 (保持不变) ---

const lettersAndNumbers = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

// generateRandomYear 生成一个有效的 YEAR 值 (1901-2155)
func generateRandomYear() int {
	return 1901 + rand.Intn(2155-1901+1) // 1901 + [0-254] => 1901-2155
}

func randomString(n int) string {
	if n <= 0 {
		return ""
	}
	sb := strings.Builder{}
	sb.Grow(n)
	for i := 0; i < n; i++ {
		sb.WriteByte(lettersAndNumbers[rand.Intn(len(lettersAndNumbers))])
	}
	return sb.String()
}

func randomChineseString(n int) string {
	if n <= 0 {
		return ""
	}

	// 使用安全的中文字符集，确保是有效的UTF-8编码
	safeChineseChars := []rune{
		'你', '好', '世', '界', '中', '国', '人', '民',
		'北', '京', '上', '海', '广', '州', '深', '圳',
		'天', '地', '山', '水', '风', '云', '日', '月',
		'星', '辰', '大', '小', '多', '少', '高', '低',
		'东', '西', '南', '北', '春', '夏', '秋', '冬',
		'红', '橙', '黄', '绿', '青', '蓝', '紫', '白',
		'黑', '金', '银', '铜', '铁', '木', '水', '火',
		'土', '光', '影', '声', '音', '色', '香', '味',
	}

	sb := strings.Builder{}
	sb.Grow(n * 3) // 中文通常占3字节，预留空间

	for i := 0; i < n; i++ {
		sb.WriteString(string(safeChineseChars[rand.Intn(len(safeChineseChars))]))
	}
	return sb.String()
}

func randomChineseName() string {
	return surnames[rand.Intn(len(surnames))] + givenNames[rand.Intn(len(givenNames))]
}

func randomBytes(n int) []byte {
	bytes := make([]byte, n)
	for i := 0; i < n; i++ {
		bytes[i] = byte(rand.Intn(256))
	}
	return bytes
}

// generateValidTimeString 生成有效的MySQL时间格式字符串
func generateValidTimeString() string {
	// 生成有效的时间格式：HH:MM:SS
	h := rand.Intn(24)
	m := rand.Intn(60)
	s := rand.Intn(60)
	return fmt.Sprintf("%02d:%02d:%02d", h, m, s)
}

func randomNullableInt(nullProb float64) interface{} {
	if rand.Float64() < nullProb {
		return nil
	}
	return rand.Intn(10000)
}

func randomNullableString(nullProb float64, maxLength int) interface{} {
	if rand.Float64() < nullProb {
		return nil
	}
	length := rand.Intn(maxLength) + 1
	return randomString(length)
}

func randomNullableTime(nullProb float64) interface{} {
	if rand.Float64() < nullProb {
		return nil
	}
	maxDurationNanos := int64(365*24) * int64(time.Hour)
	randomOffsetNanos := rand.Int63n(maxDurationNanos)
	return time.Now().Add(-time.Duration(randomOffsetNanos))
}

func randomNullableBytes(nullProb float64, maxLength int) interface{} {
	if rand.Float64() < nullProb {
		return nil
	}
	length := rand.Intn(maxLength) + 1
	return randomBytes(length)
}

// generateTypesRow 为 checksum_test_types 表生成单行数据
// ... (保持不变，已在上面提供) ...
func generateTypesRow(index int, generationType string, totalBaseRowCount int) map[string]interface{} {
	row := make(map[string]interface{})

	if generationType == "source" || generationType == "modified" {
		row["id"] = index + 1
	} else { // "extra" 类型
		row["id"] = totalBaseRowCount + index + 1 // 确保额外行的 ID 不与源库 ID 冲突
	}

	row["bool_col"] = rand.Intn(2) == 1
	row["tiny_col"] = int8(rand.Intn(256) - 128)
	row["tiny_unsigned_col"] = uint8(rand.Intn(256)) // Fix: Ensure tiny_unsigned_col stays within valid range (0-255)
	row["small_col"] = int16(rand.Intn(65536) - 32768)
	row["small_unsigned_col"] = uint16(rand.Intn(65536))
	row["medium_col"] = int32(rand.Int63n(16777216) - 8388608)
	row["medium_unsigned_col"] = uint32(rand.Int63n(16777216))
	row["int_col"] = int32(rand.Int63n(2147483648)) // -2147483648 to 2147483647
	row["int_unsigned_col"] = uint32(rand.Uint32())
	row["big_col"] = rand.Int63()           // -9223372036854775808 to 9223372036854775807
	row["big_unsigned_col"] = rand.Uint64() // 0 to 18446744073709551615
	// 修改 decimal_col 的生成方式，确保格式正确
	decimalValue := rand.Float64() * 10000
	row["decimal_col"] = fmt.Sprintf("%.2f", decimalValue)

	row["float_col"] = rand.Float32() * 1000
	row["double_col"] = rand.Float64() * 1000000

	// 修改 char_col 字段生成方式，确保长度始终是 10
	row["char_col"] = func() string {
		s := randomString(10)
		if len(s) > 10 {
			s = s[:10]
		} else if len(s) < 10 {
			s = s + strings.Repeat(" ", 10-len(s))
		}
		return s
	}()

	row["varchar_col"] = randomString(rand.Intn(50) + 10)
	row["tinytext_col"] = randomChineseString(rand.Intn(20) + 5)
	row["text_col"] = randomChineseString(rand.Intn(100) + 10)
	row["mediumtext_col"] = randomString(rand.Intn(1000) + 200)
	row["longtext_col"] = randomChineseString(rand.Intn(500) + 100)

	row["binary_col"] = randomBytes(16)
	row["varbinary_col"] = randomBytes(rand.Intn(50))
	row["tinyblob_col"] = randomBytes(rand.Intn(100))
	row["blob_col"] = randomBytes(rand.Intn(500))
	row["mediumblob_col"] = randomBytes(rand.Intn(1000))
	row["longblob_col"] = randomBytes(int(rand.Int63n(2000)))

	baseTime := time.Date(2000+rand.Intn(30), time.Month(rand.Intn(12)+1), rand.Intn(28)+1, rand.Intn(24), rand.Intn(60), rand.Intn(60), rand.Intn(1e9), time.UTC)
	row["date_col"] = baseTime
	// 使用安全的时间字符串生成函数
	row["time_col"] = generateValidTimeString()
	row["datetime_col"] = baseTime
	tsTime := time.Date(1970+rand.Intn(70), time.Month(rand.Intn(12)+1), rand.Intn(28)+1, rand.Intn(24), rand.Intn(60), rand.Intn(60), rand.Intn(1e9), time.UTC)
	row["datetime_col_was_ts"] = tsTime

	row["year_col"] = generateRandomYear()

	row["enum_col"] = []string{"apple", "banana", "orange"}[rand.Intn(3)]
	setValues := []string{"red", "green", "blue"}
	randSet := make([]string, 0)
	if rand.Intn(2) == 1 {
		randSet = append(randSet, setValues[0])
	}
	if rand.Intn(2) == 1 {
		randSet = append(randSet, setValues[1])
	}
	if rand.Intn(2) == 1 {
		randSet = append(randSet, setValues[2])
	}
	// Fix: Ensure we never have an empty SET value - if no values were selected, use at least one
	if len(randSet) == 0 {
		randSet = append(randSet, setValues[rand.Intn(len(setValues))])
	}
	row["set_col"] = strings.Join(randSet, ",")

	// 改进 JSON 数据生成，确保生成有效的 JSON
	jsonObj := map[string]interface{}{
		"name":      randomString(8),
		"active":    rand.Intn(2) == 1,
		"score":     rand.Float64() * 100,
		"id":        index + 1,
		"timestamp": time.Now().Format(time.RFC3339),
		"city":      randomChineseString(2),
		"address":   fmt.Sprintf("%s路%d号", randomChineseString(2), rand.Intn(200)+1),
	}

	jsonBytes, err := json.Marshal(jsonObj)
	if err != nil {
		log.Printf("警告: JSON 编码失败: %v，将生成简单 JSON", err)
		row["json_col"] = `{"id":1,"data":true}`
	} else {
		row["json_col"] = string(jsonBytes)
	}

	row["nullable_int"] = randomNullableInt(cfg.NULLProbability)
	row["nullable_varchar"] = randomNullableString(cfg.NULLProbability, 30)
	row["nullable_text"] = randomNullableString(cfg.NULLProbability, 100)
	row["nullable_datetime"] = randomNullableTime(cfg.NULLProbability)
	row["nullable_blob"] = randomNullableBytes(cfg.NULLProbability, 100)

	if generationType == "modified" {
		numColsToModify := rand.Intn(5) + 1
		modifiedCols := 0
		nonKeyCols := make([]string, 0, len(row))
		for colName := range row {
			if colName != "id" {
				nonKeyCols = append(nonKeyCols, colName)
			}
		}
		rand.Shuffle(len(nonKeyCols), func(i, j int) {
			nonKeyCols[i], nonKeyCols[j] = nonKeyCols[j], nonKeyCols[i]
		})

		for i := 0; i < len(nonKeyCols) && modifiedCols < numColsToModify; i++ {
			colName := nonKeyCols[i]
			row[colName] = generateValueForColumn(colName, row[colName])
			modifiedCols++
		}
	}

	return row
}

// generateCompositeRow 仅用于非 composite_key 表，移除 user_id/item_id 派生逻辑
func generateCompositeRow(index int, generationType string, totalBaseRowCount int) map[string]interface{} {
	row := make(map[string]interface{})
	// 这里只生成非 composite_key 表的其它字段
	row["description"] = randomChineseString(10)
	row["quantity"] = rand.Intn(100) + 1
	row["last_updated"] = time.Now().Add(time.Duration(rand.Intn(24)) * time.Hour)
	return row
}

// generateCompositeRowWithKey 为 checksum_test_composite_key 表生成单行数据
// 修复：确保 (user_id, item_id) 组合全局唯一，避免主键冲突
func generateCompositeRowWithKey(userID, itemID uint32, generationType string, totalBaseRowCount int) map[string]interface{} {
	row := make(map[string]interface{})
	row["user_id"] = userID
	row["item_id"] = itemID
	if generationType == "extra" {
		// 避免和源库冲突，user_id 增加偏移
		userID += 100000
		row["user_id"] = userID
	}
	// 其余字段保持 generateCompositeRow 原逻辑
	row["description"] = randomChineseString(10)
	row["quantity"] = rand.Intn(100) + 1
	row["last_updated"] = time.Now().Add(time.Duration(rand.Intn(24)) * time.Hour)
	return row
}

// 修改 generateValueForColumn 函数，特别处理 decimal_col 列的长度限制
func generateValueForColumn(colName string, sampleValue interface{}) interface{} {
	// 添加日志，记录正在处理的列和原始值类型
	// log.Printf("处理列 %s，原始值类型: %T, 值: %v", colName, sampleValue, sampleValue)

	// 如果原始值是 nil，根据列名推断适当的类型
	if sampleValue == nil {
		switch {
		case strings.Contains(colName, "int") || strings.Contains(colName, "num"):
			return rand.Int31n(1000)

		case strings.Contains(colName, "float") || strings.Contains(colName, "double") ||
			strings.Contains(colName, "decimal"):
			return fmt.Sprintf("%.2f", rand.Float64()*10000)

		case strings.Contains(colName, "date") || strings.Contains(colName, "time"):
			// 对TIME类型特殊处理，确保格式正确
			if colName == "time_col" {
				// 使用专用函数生成有效的时间字符串
				return generateValidTimeString()
			}
			return time.Now().Add(time.Duration(rand.Intn(24)) * time.Hour)

		case strings.Contains(colName, "char") || strings.Contains(colName, "text") ||
			strings.Contains(colName, "varchar"):
			// 针对字符串类型，对CHAR类型特别处理
			if colName == "char_col" {
				// char_col 在表中定义为 CHAR(10)，长度固定为10
				return randomString(5) + strings.Repeat(" ", 5) // 确保总长度为10
			}
			return randomString(rand.Intn(20) + 5)

		case strings.Contains(colName, "blob") || strings.Contains(colName, "binary"):
			return randomBytes(rand.Intn(50) + 5)

		case strings.Contains(colName, "bool"):
			return rand.Intn(2) == 1

		case strings.Contains(colName, "enum"):
			return []string{"apple", "banana", "orange"}[rand.Intn(3)]

		case strings.Contains(colName, "set"):
			setValues := []string{"red", "green", "blue"}
			randSet := make([]string, 0)
			// 随机选择0-3个值
			for _, val := range setValues {
				if rand.Intn(2) == 1 {
					randSet = append(randSet, val)
				}
			}
			// 确保至少有一个值
			if len(randSet) == 0 {
				randSet = append(randSet, setValues[rand.Intn(len(setValues))])
			}
			return strings.Join(randSet, ",")
		case strings.Contains(colName, "year"):
			return 1901 + rand.Intn(2155-1901+1)
		case strings.Contains(colName, "json"):
			// 对于JSON列，确保生成有效的JSON
			jsonObj := map[string]interface{}{
				"id":        rand.Intn(10000) + 1,
				"name":      randomString(8),
				"active":    rand.Intn(2) == 1,
				"score":     rand.Float64() * 100,
				"timestamp": time.Now().Format(time.RFC3339),
			}
			jsonBytes, err := json.Marshal(jsonObj)
			if err != nil {
				log.Printf("警告: JSON 编码失败: %v，将生成简单 JSON", err)
				return `{"id":1,"data":true}`
			}
			return string(jsonBytes)

		default:
			// 对于未知的类型，记录警告并返回原值
			log.Printf("警告: 遇到未在 generateValueForColumn 中处理的类型 %T，不修改列 '%s'", sampleValue, colName)
			return sampleValue
		}
	}

	// 根据原始值的类型生成新值
	switch v := sampleValue.(type) {
	case bool:
		// 对布尔类型，生成随机布尔值
		return rand.Intn(2) == 1

	case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
		// 特殊处理year_col列，确保值在有效范围内
		if colName == "year_col" {
			return 1901 + rand.Intn(2155-1901+1) // 1901 + [0-254] => 1901-2155
		}

		// 对整数类型，根据列名生成适当范围内的随机整数
		switch {
		case colName == "tiny_col":
			return int8(rand.Intn(256) - 128) // -128 to 127
		case colName == "tiny_unsigned_col":
			return uint8(rand.Intn(256)) // 0 to 255
		case colName == "small_col":
			return int16(rand.Intn(65536) - 32768) // -32768 to 32767
		case colName == "small_unsigned_col":
			return uint16(rand.Intn(65536)) // 0 to 65535
		case colName == "medium_col":
			return int32(rand.Int63n(16777216) - 8388608) // -8388608 to 8388607
		case colName == "medium_unsigned_col":
			return uint32(rand.Int63n(16777216)) // 0 to 16777215
		case colName == "int_col":
			return int32(rand.Int63n(2147483648)) // -2147483648 to 2147483647
		case colName == "int_unsigned_col":
			return uint32(rand.Uint32())
		case colName == "big_col":
			return rand.Int63() // -9223372036854775808 to 9223372036854775807
		case colName == "big_unsigned_col":
			return rand.Uint64() // 0 to 18446744073709551615
		default:
			return rand.Int31n(1000) // Default for other integer columns
		}

	case float32, float64:
		// 对浮点类型，根据列名生成适当范围内的随机浮点数
		switch {
		case colName == "decimal_col":
			// 确保生成有效的decimal值，格式为字符串
			return fmt.Sprintf("%.2f", rand.Float64()*10000)
		case colName == "float_col":
			return rand.Float32() * 1000
		case colName == "double_col":
			return rand.Float64() * 1000000
		default:
			return rand.Float64() * 1000
		}

	case string:
		// 特殊处理decimal_col列，确保始终是有效的decimal值
		if colName == "decimal_col" {
			return fmt.Sprintf("%.2f", rand.Float64()*10000)
		}

		// 特殊处理json_col列，确保始终是有效的JSON
		if colName == "json_col" || strings.Contains(colName, "json") {
			// 对于JSON列，确保生成有效的JSON
			jsonObj := map[string]interface{}{
				"id":        rand.Intn(10000) + 1,
				"name":      randomString(8),
				"active":    rand.Intn(2) == 1,
				"score":     rand.Float64() * 100,
				"timestamp": time.Now().Format(time.RFC3339),
			}
			jsonBytes, err := json.Marshal(jsonObj)
			if err != nil {
				log.Printf("警告: JSON 编码失败: %v，将生成简单 JSON", err)
				return `{"id":1,"data":true}`
			}
			return string(jsonBytes)
		}

		// 对字符串类型，根据列名进行特殊处理
		// 为不同列类型添加长度限制
		maxLength := 255 // 默认最大长度

		// 对特殊类型进行处理
		if colName == "time_col" {
			// 使用专用函数生成有效的时间字符串
			return generateValidTimeString()
		} else if colName == "enum_col" {
			// 枚举类型，只能是预定义的值之一
			return []string{"apple", "banana", "orange"}[rand.Intn(3)]
		} else if colName == "set_col" {
			// SET类型，只能是预定义的值的组合
			setValues := []string{"red", "green", "blue"}
			randSet := make([]string, 0)
			// 随机选择0-3个值
			for _, val := range setValues {
				if rand.Intn(2) == 1 {
					randSet = append(randSet, val)
				}
			}
			// 确保至少有一个值
			if len(randSet) == 0 {
				randSet = append(randSet, setValues[rand.Intn(len(setValues))])
			}
			return strings.Join(randSet, ",")
		}

		// 根据列名推断最大长度
		switch {
		case colName == "char_col":
			maxLength = 10 // CHAR(10) 在表结构中定义为最大长度10
			// 确保生成固定长度的字符串
			newStr := randomString(maxLength)
			if len(newStr) > maxLength {
				newStr = newStr[:maxLength]
			} else if len(newStr) < maxLength {
				newStr = newStr + strings.Repeat(" ", maxLength-len(newStr))
			}
			return newStr
		case colName == "varchar_col" || colName == "nullable_varchar":
			maxLength = 255
		case colName == "tinytext_col":
			maxLength = 255
		case colName == "text_col" || colName == "nullable_text":
			maxLength = 1000
		case colName == "mediumtext_col":
			maxLength = 2000
		case colName == "longtext_col":
			maxLength = 2000 // 虽然可以更长，但为了安全限制一下
		}

		// 确保不会超过列的最大长度
		if len(v) > maxLength {
			return v[:maxLength]
		}

		// 生成一个随机字符串，长度在原始字符串长度的基础上有所变化
		newLength := int(float64(len(v)) * (1 + rand.Float64()*0.2 - 0.1)) // 长度变化范围约 +/- 10%
		if newLength < 1 {                                                 // 确保长度至少为 1
			newLength = 1
		}
		if newLength > maxLength { // 避免超过列的最大长度
			newLength = maxLength
		}
		return randomString(newLength)

	case []byte:
		// 对二进制类型，生成随机二进制数据
		maxSize := 2000 // 默认最大大小

		// 根据列名推断最大大小
		switch {
		case colName == "binary_col":
			maxSize = 16 // 固定大小
		case colName == "varbinary_col":
			maxSize = 255
		case colName == "tinyblob_col":
			maxSize = 255
		case colName == "blob_col" || colName == "nullable_blob":
			maxSize = 1000
		case colName == "mediumblob_col":
			maxSize = 2000
		case colName == "longblob_col":
			maxSize = 2000
		}

		// 确保不会超过列的最大大小
		if len(v) > maxSize {
			return v[:maxSize]
		}

		// 生成随机二进制数据，大小在原始数据大小的基础上有所变化
		newSize := int(float64(len(v)) * (1 + rand.Float64()*0.2 - 0.1)) // 大小变化范围约 +/- 10%
		if newSize < 1 {                                                 // 确保大小至少为 1
			newSize = 1
		}
		if newSize > maxSize { // 避免超过列的最大大小
			newSize = maxSize
		}
		return randomBytes(newSize)

	case time.Time:
		// 对时间类型，生成一个随机偏移的时间
		maxOffsetHours := 24 * 30 // 最大偏移约为一个月
		offsetHours := rand.Intn(maxOffsetHours*2) - maxOffsetHours
		return v.Add(time.Duration(offsetHours) * time.Hour)

	default:
		// 对于未知的类型，记录警告并返回原值
		log.Printf("警告: 遇到未在 generateValueForColumn 中处理的类型 %T，不修改列 '%s'", sampleValue, colName)
		return sampleValue
	}
}
